package zero.tech.redis.delay;

import java.util.concurrent.TimeUnit;

import org.redisson.api.RDelayedQueue;
import org.redisson.api.RPriorityBlockingQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class RedisDelayedQueueUtil {
	@Autowired(required = false)
    private RedissonClient redissonClient;
	
	public void addQueue(RedisDelayQueueType type,RedisDelayedEntity t) {
       addQueue(type.getQueue(), t);
    }
	
	public void addQueue(String queue,RedisDelayedEntity t) {
		RPriorityBlockingQueue<RedisDelayedEntity> blockingFairQueue = redissonClient.getPriorityBlockingQueue(queue);
        RDelayedQueue<RedisDelayedEntity> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
        if(t.getPush_time()==t.getExec_time()) {
        	delayedQueue.offer(t, 0, TimeUnit.SECONDS);
        }else {
        	delayedQueue.offer(t, t.getDelay(), TimeUnit.SECONDS);
        }
//        delayedQueue.destroy();
    }
	public <T>void getQueue(RedisDelayQueueType queue, RedisDelayedEvent<T> taskEventListener) {
		getQueue(queue.getQueue(), taskEventListener);
	}
	
	public <T>void getQueue(String queue, RedisDelayedEvent<T> taskEventListener) {
		RPriorityBlockingQueue<? extends RedisDelayedEntity> blockingFairQueue = redissonClient.getPriorityBlockingQueue(queue);
		RDelayedQueue<? extends RedisDelayedEntity> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
		System.out.println(delayedQueue.isExists());
        //由于此线程需要常驻，可以新建线程，不用交给线程池管理
        new Thread(new Runnable() {
			@SuppressWarnings("unchecked")
			@Override
			public void run() {
				while (true) {
	                try {
	                	RedisDelayedEntity redisDelayedEntity = blockingFairQueue.take();
	                	T t=(T) redisDelayedEntity;
	                    taskEventListener.invoke(t);
	                } catch (Exception e) {
	                    e.printStackTrace();
	                }
	            }
			}
		}).start();
    }
}
